package com.aotain.nyx.port;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

import com.aotain.apollo.ApolloConfig;
import com.aotain.apollo.PortInfo;
import com.aotain.common.mongo.DataColumn;
import com.aotain.apollo.mongodb.MongoImportTool;
import com.aotain.apollo.utils.ApolloProperties;
import com.aotain.nyx.flink.BasicTuple;
import com.aotain.nyx.statis.AbnStatisTuple;

/**
 * 
     *@Title:端口异常检测
     *@Description:
     *@Author:Turk
     *@Since:2017年6月30日
     *@Version:1.1.0
 */
public class PortDetect implements FlatMapFunction<Tuple2<String,BasicTuple>, AbnStatisTuple>{

	
	/**
	  *
	  */
	private static final long serialVersionUID = 7506117984941846192L;

	HashMap<Integer,PortInfo> ports = null;
	
	public PortDetect(String dbJson) {
		//初始化
		ApolloConfig ap = new ApolloConfig(dbJson);
				
		//加载配置信息
		ports = ap.getPortThreaten();
				
		ports.put(22,new PortInfo(22, 80, "SSH远程协议访问",0));
		ports.put(21,new PortInfo(21, 80, "ftp访问",0));
		ports.put(23,new PortInfo(23, 80, "Telnet访问",0));
		ports.put(3389, new PortInfo(3389, 80, "Windows远程终端访问",0));
		//135、139、445、593、1025
		ports.put(135, new PortInfo(135, 80, "Windows DCOM服务访问",0));
		ports.put(139, new PortInfo(139, 80, "NetBIOS/SMB服务访问",0));
		ports.put(445, new PortInfo(445, 80, "公共Internet文件系统访问",0));
		ports.put(1433, new PortInfo(1433, 80, "SQL Server常用端口访问",0));
	}

	@Override
	public void flatMap(Tuple2<String, BasicTuple> value,
			Collector<AbnStatisTuple> out) throws Exception {
		
		BasicTuple tuple = value.f1;
		
		PortInfo portinfo = ports.get(Integer.parseInt(tuple.getDestPort()));
		
		String destip = tuple.getDestIP();
		String sourceip = tuple.getSourceIP();
		String destport = tuple.getDestPort();
		String gisstr = tuple.getGIS();
		
		
	
		String desc = "";
		int score = 70;			
		if(portinfo!=null)
		{
			desc = desc + "" +portinfo.Desc;
			score = portinfo.score;
		}		
		
		String[] gisArray = gisstr.split("#");
		
		//Logger.getLogger(PortDetectBolt.class).info("LANG DEBUG="+gisstr);
		
		String destAreaName = gisArray[0]; //如果地市为空，精确到国家
		String DestGeo = gisArray[1];
		String sourceAreaName = gisArray[2];  //如果地市为空，精确到国家
		String SourceGeo = gisArray[3];
		String sourceAreaCountry = gisArray[4];
		String sourceProvinceID =  gisArray[5];
		String sourceProvinceName = gisArray[6].isEmpty()?sourceAreaCountry:gisArray[6]; //如果省为空，精确到国家
		String sourceAreaCityId = gisArray[7];
		String sourceAreaProvinceId = gisArray[8];
		
		Long AttNum = tuple.getSessionNum();
		

		Date current = new Date();
		SimpleDateFormat sdf  = new SimpleDateFormat("yyyyMMddHHmmss");
		String strDate = sdf.format(current);
		
		String rowkey = String.format("%s_%s_%s_%s_%d", destip, strDate, 
				sourceip,destport,4);
		
    	//##########COPY LOG
    	MongoImportTool importtool = MongoImportTool.getInstance();
		String tableName = "SDS_ABNORMAL_LOG2";
		List<DataColumn> row = new ArrayList<DataColumn>();
		
		row.add(new DataColumn("ROWKEY",rowkey)); //主键字段，用于更新记录
		row.add(new DataColumn("SOURCEIP",sourceip));
		row.add(new DataColumn("DESTPORT",destport));
		row.add(new DataColumn("ACCESSTIME",Long.parseLong(strDate)));
		row.add(new DataColumn("ENDTIME",Long.parseLong(strDate)));
		row.add(new DataColumn("ABRNORMAL", 4));
		row.add(new DataColumn("DESTIP", destip));
		row.add(new DataColumn("SOURCEAREA", sourceAreaName));
		row.add(new DataColumn("SOURCEGEO",  String.format("%s",SourceGeo)));
		row.add(new DataColumn("SOURCECOUNTRY", sourceAreaCountry));
		row.add(new DataColumn("DESTAREA", destAreaName));
		row.add(new DataColumn("DESTGEO",  String.format("%s",DestGeo)));
		row.add(new DataColumn("ATTNUM", AttNum));
		desc = String.format("%s 发生连接%d次", desc , AttNum);
		row.add(new DataColumn("DESC", desc));
		row.add(new DataColumn("EVALUATE", String.valueOf(score)));
		row.add(new DataColumn("PROVINCE", sourceProvinceName));
		
		row.add(new DataColumn("UPSTREAMOCTETS", tuple.getUpStreamOctets()));
		row.add(new DataColumn("UPSTREAMPACKET", tuple.getUpStreamPacket()));
		row.add(new DataColumn("ATTTYPE", "2"));
		row.add(new DataColumn("FLOWDIRECTION", tuple.getFlowDirection()));

		importtool.InsertRowData(tableName, row);

		AbnStatisTuple retTupleOct = new AbnStatisTuple();
		retTupleOct.setDestIP(destip);
		retTupleOct.setSourceIP(sourceip);
		retTupleOct.setDestPort(destport);
		retTupleOct.setType("PORTLOW");
		retTupleOct.setUpStreamOctets(tuple.getUpStreamOctets());
		retTupleOct.setUpStreamPacket(tuple.getUpStreamPacket());
		retTupleOct.setAttackNum(AttNum);
		retTupleOct.setSessionNum(AttNum);
		retTupleOct.setSourceCountry(sourceAreaCountry);
		retTupleOct.setSourceProvince(sourceProvinceName);
		retTupleOct.setAttackType("2");
		retTupleOct.setAttTime(strDate);
		out.collect(retTupleOct);
		
	
	}

}

    